package app.dwd;

import app.BaseSQLApp;
import common.Constant;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import utils.SQLUtil;

public class Dwd_15_ExamQuestionDetail extends BaseSQLApp {
    public static void main(String[] args) {
        new Dwd_15_ExamQuestionDetail().init(
                5002,
                2,
                "Dwd_15_CourseExamDetail"
        );
    }

    @Override
    protected void handle(StreamExecutionEnvironment env, StreamTableEnvironment tEnv) {
        //从 ods_db 读数据
        readOdsDb(tEnv,"Dwd_15_ExamPaperDetail");
        // 从 ods_db 过滤
        Table table = tEnv.sqlQuery(" select " +
                " data['id'] id," +
                " data['exam_id'] exam_id," +
                " data['paper_id'] paper_id," +
                " cast(data['question_id'] as bigint) question_id," +
                " data['user_id'] user_id," +
                " data['score'] score," +
                " data['answer'] answer," +
                " data['is_correct'] is_correct," +
                " data['create_time'] create_time," +
                " pt" +
                " from ods_db" +
                " where `database`='edu'" +
                " and `table`='test_exam_question'");
        tEnv.createTemporaryView("test_exam_question",table);

        // 从 mysql 读取 test_question_info 维度表
        tEnv.executeSql(" create table test_question_info(" +
                " id bigint," +
                " question_txt string," +
                " chapter_id bigint," +
                " question_type bigint" +
                ")with( " +
                " 'connector'='jdbc'," +
                " 'driver'='com.mysql.cj.jdbc.Driver'," +
                " 'url' = 'jdbc:mysql://hadoop102:3306/edu?useSSL=false'," +
                " 'table-name' = 'test_question_info'," +
                " 'username' = 'root'," +
                " 'password' = '123321'," +
                " 'lookup.cache.max-rows' = '10'," +
                " 'lookup.cache.ttl' = '1 hour' " +
                " )");

        //将test_exam_question与 test_question_info 进行look up join 补充维度
        Table result = tEnv.sqlQuery(" select" +
                " teq.id," +
                " teq.exam_id," +
                " teq.paper_id," +
                " teq.question_id," +
                " tqi.question_txt," +
                " teq.user_id," +
                " teq.answer," +
                " teq.score," +
                " teq.is_correct," +
                " teq.create_time " +
                " from test_exam_question teq" +
                " join test_question_info for system_time as of teq.pt as tqi " +
                " on teq.question_id=tqi.id");


        // 在kafka 创建dwd_exam_QuestionDetail 表
        tEnv.executeSql(" create table dwd_exam_question_detail(" +
                " id string," +
                " exam_id string," +
                " paper_id string," +
                " question_id bigint," +
                " question_txt string," +
                " user_id string," +
                " answer string," +
                " score string," +
                " is_correct string," +
                " create_time string," +
                " primary key(id)not enforced" +
                ")" + SQLUtil.getKafkaSink(Constant.DWD_EXAM_QUESTION_DETAIL));
        result.executeInsert("dwd_exam_question_detail");

    }
}